package com.yds.flow

import android.os.Bundle
import androidx.activity.enableEdgeToEdge
import androidx.appcompat.app.AppCompatActivity
import androidx.core.view.ViewCompat
import androidx.core.view.WindowInsetsCompat
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collectIndexed
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.merge
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onEmpty
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.onSubscription
import kotlinx.coroutines.flow.retry
import kotlinx.coroutines.flow.retryWhen
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.io.IOException

/**
 * 作者:     YDS
 * 创建时间: 2024/1/3 12:02
 * 页面描述: Flow
 *
 * 回调操作符：
 * 1.onStart 在上游发送数据开始之前调用-单次
 * 2.onEach 在上游想下游发送数据之前调用-多次
 * 3.onEmpty 当flow完成时却没有发送任何数据时调用，可以用来兜底 onEmpty(emit("data"))
 * 4.onSubscription SharedFlow/StateFlow专属操作符，多用于发送数据
 * 5.onCompletion 在flow取消或者完成时调用
 *
 * 变换操作符：
 * 1.map 将上流发送的数据惊醒变换操作
 * 2.mapLatest 当有新值发送时，上个数据变换还未结束会立即取消掉发送当前数据
 * 3.mapNotNull 仅发送map后不为null的值
 * 4.asStateFlow asSharedFlow 将MutableStateFlow与MutableSharedFlow转换为不可变的Flow用来对外暴露数据
 * 5.shareIn StateIn 将flow转换为SharedFlow与StateFlow（未尝试过）
 *
 * 过滤操作符：
 * 1.filter 筛选出符合条件的值
 * 2.filterInstance 筛选对应类型的值
 * 3.filterNot 筛选出不符合条件的值
 * 4.filterNotNull 筛选出不为空的值
 * 5.drop 丢弃掉数据前N个值
 * 6.dropWhile 找到第一个不满足条件的值，返回其和其后所有的值
 * 7.take 返回前N个元素
 * 8.takeWhile 找到第一个不满足条件的值，返回之前的全部数据
 * 9.debounce 节流防抖 debounce(1000)返回1000毫秒内最后一次接收的值
 * 10.sample 采样 sample(1000)仅获取周期1000毫秒内最新发出的值
 *
 * 组合操作符：
 * 1.combine 组合每个流最新发送的值
 * 2.zip 组合每个流发送的值，一旦一个流结束了，那么整个过程也就结束了
 * 3.merge
 * @see FlowActivity.merge 合并多个流为一个流
 *
 * 功能性操作符：
 * 1.cancellable 接收的时候判断携程是否取消，如果已取消报错抛异常
 * 2.catch 对上游进行异常捕获，上游指的是此操作符之前的流，下游指的是此操作符之后的流
 * 3.retryWhen
 * @see FlowActivity.retryWhen 有条件的进行重试，lambda中有两个参数，第一个是异常原因，第二个是重试的次数，返回值为boolean，true为继续重试，false为结束重试
 * 4.retry
 * @see FlowActivity.retry 重试机制，retry(n) n最大尝试次数
 * 5.buffer
 * @see FlowActivity.buffer 缓冲区，多用于解决背压问题
 * 6.flowOn 指定上游操作的执行线程
 *
 * 末端操作符：
 * 1.collect 触发flow运行，常用的结果监听方式
 * 2.collectIndexed 带下标的触发flow运行
 * 3.collectLatest 如果上流发出数据，下流未接收完成，上流再次发送数据会取消掉上个数据的接收操作
 * 4.toCollection toList toSet 将结果添加到集合
 * 5.launchIn 指定下游操作的执行线程，直接触发flow的操作，入参为coroutineScope，搭配OnEach或onCompletion使用，返回值为job
 * 6.lash 返回上流发送的最后一个值，如果为null会抛异常
 * 7.lashOrNull 返回上流发送的最后一个值，可以为null
 * 8.first 返回上流发送的第一个值，如果为null会抛异常
 * 9.firstOrNull  返回上流发送的第一个值，可以为null
 * 10.single 接收流发送的第一个值，区别于first，如果为空或者发了不止一个值都会报错
 * 11.singleOrNull 接收流发送的第一个值，可以为空，如果发送多个值仅取第一个值
 * 12.count 返回上流发送值的个数
 * 13.fold 需设置初始值，执行遍历，并将结果作为下次执行的参数fold(0,{result,value->})
 * 14.reduce 无需设置初始值，执行遍历，并将结果作为下次执行的参数reduce{result,value->}
 * 15.distinctUntilChangedBy 去重操作符，判断连续两个值是否重复
 * 16.distinctUntilChanged 用于过滤连续两个值一样，跳过发送动作
 *
 */
class FlowActivity : AppCompatActivity() {
    private val mListDigit = mutableListOf(1, 2, 4, 4, 5, 3)
    private val mListWord = mutableListOf("a", "b", "c", "d")
    private val mListMix = mutableListOf("a", "b", 1, 2)

    private lateinit var job: Job

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        enableEdgeToEdge()
        setContentView(R.layout.activity_flow)

        retryWhen()
        retry()
        buffer()
        merge()

//        coldFlow()
//        hotFlow()

        countDown()
    }

    private fun countDown() {
        runBlocking {
            flow {
                (10 downTo 0).forEach {
                    delay(1000)
                    emit(it)
                }
            }.onStart {
                println("倒计时开始")
            }
                .onCompletion {
                    println("倒计时结束")
                }.catch {
                    kotlin.io.println("catch: ${it.message}")
                }.collect {
                    // 在这里 更新值来显示到UI
                    println("倒计时：$it")
                }
        }
    }

    /*** 作者: YDS 时间: 2024/1/3 11:37 描述: 冷流->必须执行末端中止操作符如collect single first toList等才会执行 ***/
    private fun coldFlow() {
        runBlocking {
            // asFlow 将集合中的元素转换为固定元素的Flow
            mListMix.asFlow()
                .filterIsInstance<String>() // 筛选对应类型的值
                .collect {
                    println("asFlow: $it")
                }

            // flowOf 创建一个包含固定元素的Flow
            flowOf(1, 2, 3).collect {
                println("flowOf: $it")
            }

            // flow channelFlow 切换线程及中间操作符
            mListDigit.asFlow()
                .flowOn(Dispatchers.IO)// flowOn操作符用于切换上游的上下文
                .onStart {// 发送数据之前回调
                    println("onStart")
                }
                .onEach {// 发送数据之后回调
                    println("onEach: $it")
                }
                .onEmpty {
                    println("onEmpty:")
                }
                .onCompletion {
                    println("onCompletion")
                }
                .map {// 对发送数据进行操作
                    it * 2
                }
                .filter {// 对发送数据进行过滤
                    it >= 3
                }
//                .take(2)// 限长操作符
//                .launchIn(this)// launchIn操作符用于切换下游的上下文
                .distinctUntilChanged()// 去除重复发送的数据
                .catch {
                    println("catch: ${it.message}")
                }
//                .collect {// 接收数据
//                    println("collect: $it")
//                    mTvMessage.text = it.toString()
//                }
//                .collectLatest {// 与 collect的区别是，有新值发出时，如果此时上个收集尚未完成，则会取消掉上个值的收集操作。
//                    println("collectLatest: $it")
//                }
                .collectIndexed { index, value ->// 带下标的 接收数据
                    println("collectIndexed: $index  $value ")
                }

            // 空流
            emptyFlow<Int>().collect { }

            /*channelFlow {
                send(1)
                withContext(Dispatchers.IO) {
                    delay(1000)
                    send(2)
                }
            }.collect {
                println(it)
            }*/

            // zip组合操作符,对两个flow数据进行组合数据处理
            val flowA = mListDigit.asFlow()
            val flowB = mListWord.asFlow()
            flowA.zip(flowB) { a, b ->
                "$a -> $b"
            }
                .collect {
                    println("zip result: $it")
                }

            // combine组合操作符,对两个flow数据进行组合数据处理
            flowA.combine(flowB) { a, b ->
                "$a -> $b"
            }
                .collect {
                    println("combine result: $it")
                }
        }
    }

    /*** 作者: YDS 时间: 2024/1/3 11:37 描述: 热流->StateFlow ShareFlow 只要数据更新Flow就会执行 ***/
    private fun hotFlow() {
        runBlocking {
            val sharedFlow = MutableSharedFlow<String>().onSubscription {
                emit("onSubscription")
            }
                .onStart { emit("onStart") }

            launch {
                sharedFlow.collect {
                    println("MutableSharedFlow collect: $it")
                }
            }

            val progressStateFlow = MutableStateFlow(0)
            job = progressStateFlow.onEach {
                println("进度更新: $it")
                if (it == 10) {
                    job.cancel()
                }
            }.launchIn(this)

            launch {
                while (isActive) {
                    delay(100)
                    progressStateFlow.value++
                }
            }
        }
    }

    private fun retryWhen() {
        runBlocking {
            flow {
                emit(1)
                delay(500)
                throw IOException("game over")
            }
                .retryWhen { cause, attempt ->
                    if (attempt > 3) {
                        return@retryWhen false
                    }
                    cause is IOException
                }
                .catch {
                    println("retryWhen catch: ${it.message}")
                }
                .collect {
                    println("retryWhen: $it")
                }
        }
    }

    private fun retry() {
        runBlocking {
            flow {
                emit(1)
                delay(500)
                throw IOException("game over")
            }
                .retry(3)
                .catch {
                    println("retry catch: ${it.message}")
                }
                .collect {
                    println("retry: $it")
                }
        }
    }

    /*** 作者: YDS 时间: 2024/1/3 15:27 描述: flow的背压处理,当上游的发送速度大于下游的接收速度会出现等待现象,可以通过buffer处理. ***/
    private fun buffer() {
        runBlocking {
            val startTime = System.currentTimeMillis()
            flow {
                emit(1)
                delay(1000)
                emit(2)
                delay(1000)
                emit(3)
            }
                .buffer(100)
                .collect {
                    delay(2000)
                    println("背压：$it")
                }
            println("背压耗时：${System.currentTimeMillis() - startTime}")
        }
    }

    /*** 作者: YDS 时间: 2024/1/3 15:55 描述: 多流合单流 ***/
    private fun merge() {
        runBlocking {
            listOf(mListDigit.asFlow(), mListWord.asFlow())
                .merge()
                .collect {
                    println("merge：$it")
                }
        }
    }

}